package com.cmbyc.source

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}

/**
 *
 * @program: com.cmbyc.source
 * @author: YCLW058
 * @create: 2021-05-18 14:40
 * @decsription:
 *
 * */

// 定义样例类，温度传感器
case class SensorReading(id: String, timestamp: Long, temperature: Double)

object Demo01 {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 1. 从集合中读取数据
    val dataList = List(
      SensorReading("sensor_1", 1547718199, 35.8),
      SensorReading("sensor_6", 1547718201, 15.4),
      SensorReading("sensor_7", 1547718202, 6.7),
      SensorReading("sensor_10", 1547718205, 38.1)
    )
    val stream1 = env.fromCollection(dataList)
    //    env.fromElements(1.0, 35, "hello")

    // 2. 从文件中读取数据
    //val inputPath = "D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt"
    //val stream2 = env.readTextFile(inputPath)

    // 3. 从kafka中读取数据

    // 4. 自定义Source
    stream1.print()

    // 执行
    env.execute("source test")
  }
}

//// 自定义SourceFunction
//class MySensorSource() extends SourceFunction[SensorReading]{
//  // 定义一个标识位flag，用来表示数据源是否正常运行发出数据
//  var running: Boolean = true
//
//  override def cancel(): Unit = running = false
//
//  override def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {
//    // 定义一个随机数发生器
//    val rand = new Random()
//
//    // 随机生成一组（10个）传感器的初始温度: （id，temp）
//    var curTemp = 1.to(10).map( i => ("sensor_" + i, rand.nextDouble() * 100) )
//
//    // 定义无限循环，不停地产生数据，除非被cancel
//    while(running){
//      // 在上次数据基础上微调，更新温度值
//      curTemp = curTemp.map(
//        data => (data._1, data._2 + rand.nextGaussian())
//      )
//      // 获取当前时间戳，加入到数据中，调用ctx.collect发出数据
//      val curTime = System.currentTimeMillis()
//      curTemp.foreach(
//        data => ctx.collect(SensorReading(data._1, curTime, data._2))
//      )
//      // 间隔500ms
//      Thread.sleep(500)
//    }
//  }


